package com.project.producer.controller;

import com.alibaba.fastjson.JSONObject;
import com.project.producer.common.ResponseMsg;
import com.project.producer.listener.SendCallbackListener;
import com.project.producer.modal.OrderStep;
import com.project.producer.utils.ListSplitter;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;

@RestController
@RequestMapping("/mqMessage")
public class MqMessageController {

    private static final Logger log = LoggerFactory.getLogger(MqMessageController.class);

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.sync-tag}")
    private String syncTag;

    @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.async-tag}")
    private String asyncag;

    @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.oneway-tag}")
    private String onewayTag;

    /**
     * 同步消息
     *
     * @param id
     * @return
     */
    @RequestMapping("/pushMessage")
    public ResponseMsg pushMessage(@RequestParam("id") int id) {

        log.info("pushMessage start : " + id);
        String messageStr = "order id : " + id;
        Message<String> message = MessageBuilder.withPayload(messageStr).setHeader(RocketMQHeaders.KEYS, id).build();
        SendResult result = rocketMQTemplate.syncSend(syncTag, message);
        log.info("pushMessage finish : " + id + ", sendResult : " + JSONObject.toJSONString(result));
        ResponseMsg msg = new ResponseMsg();
        // 解析发送结果
        if (result.getSendStatus() == SendStatus.SEND_OK) {
            msg.setSuccessData(result.getMsgId() + " : " + result.getSendStatus());
        }
        return msg;
    }

    /**
     * 异步消息
     *
     * @param id
     * @return
     */
    @RequestMapping("/pushAsyncMessage")
    public ResponseMsg pushAsyncMessage(@RequestParam("id") int id) {
        log.info("pushAsyncMessage start : " + id);
        String messageStr = "order id : " + id;
        Message<String> message = MessageBuilder.withPayload(messageStr).setHeader(RocketMQHeaders.KEYS, id).build();
        rocketMQTemplate.asyncSend(asyncag, message, new SendCallbackListener(id));
        log.info("pushAsyncMessage finish : " + id);
        ResponseMsg msg = new ResponseMsg();
        msg.setSuccessData(null);
        return msg;
    }

    /**
     * 发送单向消息
     *
     * @param id
     * @return
     */
    @RequestMapping("/pushOneWayMessage")
    public ResponseMsg pushOneWayMessage(@RequestParam("id") int id) {
        log.info("pushOneWayMessage start : " + id);
        // 构建消息
        String messageStr = "order id : " + id;
        Message<String> message = MessageBuilder.withPayload(messageStr)
                .setHeader(RocketMQHeaders.KEYS, id)
                .build();
        rocketMQTemplate.sendOneWay(onewayTag, message);
        log.info("pushOneWayMessage finish : " + id);
        ResponseMsg msg = new ResponseMsg();
        msg.setSuccessData(null);
        return msg;
    }

    /**
     * 发送有顺序的单向消息
     *
     * @param id
     * @return
     */
    @RequestMapping("/pushSequeueMessage")
    public ResponseMsg pushSequeueMessage(@RequestParam("id") int id) {
        log.info("pushSequeueMessage start : " + id);
        for (int i = 0; i < 10; i++) {
            String myId = id + "" + i;
            List<OrderStep> list = OrderStep.buildOrderSteps(myId);
            for (OrderStep item : list) {
                String messageStr = String.format("order id : %s, desc : %s", item.getId(), item.getDesc());
                Message<String> message = MessageBuilder.withPayload(messageStr)
                        .setHeader(RocketMQHeaders.KEYS, item.getId())
                        .build();
                rocketMQTemplate.setMessageQueueSelector((list1, message1, o) -> {
                    int queueNum = Integer.parseInt(String.valueOf(o)) % list1.size();
                    log.info(String.format("queueNum : %s, message : %s", queueNum, new String(message1.getBody())));
                    return list1.get(queueNum);
                });
                rocketMQTemplate.syncSendOrderly(syncTag, message, item.getId());
            }
        }
        log.info("pushSequeueMessage finish : " + id);
        ResponseMsg msg = new ResponseMsg();
        msg.setSuccessData(null);
        return msg;
    }

    /**
     * rocketmq 延迟消息
     *
     * @param id 消息
     * @return 结果
     */
    @RequestMapping("/pushDelayMessage")
    public ResponseMsg pushDelayMessage(@RequestParam("id") int id) {
        log.info("pushDelayMessage start : " + id);
        // 构建消息
        String messageStr = "order id : " + id;
        Message<String> message = MessageBuilder.withPayload(messageStr)
                .setHeader(RocketMQHeaders.KEYS, id)
                .build();
        // 设置超时和延时推送
        // 超时时针对请求broker然后结果返回给product的耗时
        // 现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18
        // private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
        SendResult sendResult = rocketMQTemplate.syncSend(syncTag, message, 1 * 1000L, 4);
        log.info("pushDelayMessage finish : " + id + ", sendResult : " + JSONObject.toJSONString(sendResult));
        ResponseMsg msg = new ResponseMsg();
        // 解析发送结果
        if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
            msg.setSuccessData(sendResult.getMsgId() + " : " + sendResult.getSendStatus());
        }
        return msg;
    }

    /**
     * 同时发送10个单向消息（真正的批量）
     *
     * @param id 消息
     * @return 结果
     */
    @RequestMapping("/pushBatchMessage")
    public ResponseMsg pushBatchMessage(@RequestParam("id") int id) {
        log.info("pushBatchMessage start : " + id);
        // 创建消息集合
        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            String myId = id + "" + i;
            // 处理当前订单唯一标识
            String messageStr = "order id : " + myId;
            Message<String> message = MessageBuilder.withPayload(messageStr)
                    .setHeader(RocketMQHeaders.KEYS, myId)
                    .build();
            messages.add(message);
        }
        // 批量下发消息到broker,不支持消息顺序操作，并且对消息体有大小限制（不超过4M）
        ListSplitter splitter = new ListSplitter(messages, 1024 * 1024 * 4);
        while (splitter.hasNext()) {
            List<Message> listItem = splitter.next();
            rocketMQTemplate.syncSend(syncTag, listItem);
        }
        log.info("pushBatchMessage finish : " + id);
        ResponseMsg msg = new ResponseMsg();
        msg.setSuccessData(null);
        return msg;
    }

    /**
     * sql过滤消息
     *
     * @param id 消息
     * @return 结果
     */
    @RequestMapping("/pushSqlMessage")
    public ResponseMsg pushSqlMessage(@RequestParam("id") int id) {
        log.info("pushSqlMessage start : " + id);
        // 创建消息集合
        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            String myId = id + "" + i;
            // 处理当前订单唯一标识
            String messageStr = "order id : " + myId;
            Message<String> message = MessageBuilder.withPayload(messageStr)
                    .setHeader(RocketMQHeaders.KEYS, myId)
                    .setHeader("money", i)
                    .build();
            messages.add(message);
        }
        rocketMQTemplate.syncSend(syncTag, messages);
        log.info("pushSqlMessage finish : " + id);
        ResponseMsg msg = new ResponseMsg();
        msg.setSuccessData(null);
        return msg;
    }

    /**
     * 事务消息
     *
     * @param id 消息
     * @return 结果
     */
    @RequestMapping("/pushTransactionMessage")
    public ResponseMsg pushTransactionMessage(@RequestParam("id") int id) {
        log.info("pushTransactionMessage start : " + id);
        // 创建消息
        String messageStr = "order id : " + id;
        Message<String> message = MessageBuilder.withPayload(messageStr)
                .setHeader(RocketMQHeaders.KEYS, id)
                .setHeader("money", 10)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, id)
                .build();
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(syncTag, message, null);
        log.info("pushTransactionMessage result : " + JSONObject.toJSONString(transactionSendResult));
        log.info("pushTransactionMessage finish : " + id);
        ResponseMsg msg = new ResponseMsg();
        msg.setSuccessData(null);
        return msg;
    }

}
